package com.shujia.cep

import com.shujia.core.Demo6WaterMark2.Event2
import org.apache.flink.api.scala._
import org.apache.flink.cep.pattern.conditions.IterativeCondition
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

object Demo3CEP {
  def main(args: Array[String]): Unit = {
    /**
      *
      * 1,1574667000,a
      * 1,1574668000,a
      * 1,1574669000,a
      * 1,1574670000,b
      * 1,1574671000,b
      * 1,1574675000,b
      * 1,1574676000,b
      * 1,1574677000,a
      * 1,1574678000,b
      * 1,1574685000,b
      *
      * 如果同一个id连续两次出现了fail   报警
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //指定时间模式
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    env.setParallelism(1)
    val ds = env.socketTextStream("node1", 8888)

    val eventDS = ds.map(_.split(","))
      .map(arr => Event(arr(0).toInt, arr(1).toLong, arr(2)))
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Event](Time.seconds(5)) {
        override def extractTimestamp(element: Event): Long = {
          element.time
        }
      })

    /**
      * 定义规则
      */

    val pattern = Pattern

      .begin[Event]("start") //第一个事件
      .where(event => {
      println(event+"start")
      event.msg.equals("a")
    }) // 条件
      .next("next")
      .where(event => {
        println(event+"next")
        event.msg.equals("b")
      })
      //可以使用事件时间和处理时间
      .within(Time.seconds(5)) //在指定时间内发生


    /**
      * 在10s内完成了中加入购物车到付款的行为
      * 说明这个人对这个商品兴趣高
      *
      */

    //使用规则在流上进行匹配
    val patternStream = CEP.pattern(eventDS.keyBy(_.id), pattern)


    //如果匹配上  就可以取出匹配成功的数据
    patternStream.select(map => {
      val start = map.getOrElse("start", null)
      val next = map.getOrElse("next", null)

      println(start)
      println(next)

    })


    env.execute()


  }

  case class Event(id: Int, time: Long, msg: String)

}
